package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"bytes"
	crand "crypto/rand"
	"fmt"
	"math/big"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"

	"github.com/ackerman/mitmr/labgob"
	"github.com/ackerman/mitmr/labrpc"
)

// import "bytes"
// import "../labgob"

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to		 the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in Lab 3 you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh; at that point you can add fields to
// ApplyMsg, but set CommandValid to false for these other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
}

type Role int

const (
	Leader Role = iota
	Follower
	Candidate
	Transfer
	EXIT
)

var roleMsg []string = []string{"Leader...", "Follwer..", "Candidate", "Transfer", "EXIT"}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

	// persistent state on all servers
	currentTerm int
	votedFor    int
	log         []Log
	// volatile state on all servers
	committedIndex int
	lastApplied    int
	// volatie state on leaders
	nextIndex  []int
	matchIndex []int

	// other

	role        Role
	appendEvent chan Role
	voteEvent   chan Role
	transfer    chan Role
	jobSignal   chan Role
	clientLog   chan Log
	ApplyChan   chan ApplyMsg

	members  []sync.Mutex
	backup   []int
	matchLog []MatchLog
}

type Log struct {
	Command interface{}
	Term    int
}

type MatchLog struct {
	beg  int
	end  int
	term int
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	var term int
	var isleader bool
	// Your code here (2A).
	term = rf.currentTerm
	if rf.role == Leader {
		isleader = true
	} else {
		isleader = false
	}
	var print bool = true
	for i := 0; i < len(rf.log); i++ {
		_, ok := rf.log[i].Command.(int)
		if !ok {
			print = false
			break
		}
	}
	if print {
		fmt.Printf(" 【Get State】 me: %d, currentTerm: %d, voteFor: %d, role: %v, log= %v\n", rf.me, rf.currentTerm, rf.votedFor, roleMsg[rf.role], rf.log)
	} else {
		fmt.Printf(" 【Get State】 me: %d, currentTerm: %d, voteFor: %d, role: %v \n", rf.me, rf.currentTerm, rf.votedFor, roleMsg[rf.role])
	}
	return term, isleader
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)

	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.currentTerm)
	e.Encode(rf.votedFor)
	e.Encode(rf.log)
	data := w.Bytes()
	rf.persister.SaveRaftState(data)

	// fmt.Printf("### [%d]    persist: { currentTerm: %d, votedFor: %d, log: %v } ###\n",
	// rf.me, rf.currentTerm, rf.votedFor, rf.log)
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {

	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }

	r := bytes.NewBuffer(data)

	d := labgob.NewDecoder(r)
	var t int
	var v int
	var l []Log
	if d.Decode(&t) != nil ||
		d.Decode(&v) != nil ||
		d.Decode(&l) != nil {

		panic("decode error")
	} else {
		rf.currentTerm = t
		rf.votedFor = v
		rf.log = l
	}

	// fmt.Printf("### [%d]readPersist: { currentTerm: %d, votedFor: %d, log: %v } ###\n", rf.me, t, v, l)
}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateId  int
	LastLogTerm  int
	LastLogIndex int
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.killed() {
		return
	}

	var prevT int = rf.currentTerm
	var prevVote int = rf.votedFor
	defer func() {
		for i := 0; i < len(rf.log); i++ {
			_, ok := rf.log[i].Command.(int)
			if !ok {
				return
			}
		}
		if reply.VoteGranted {
			fmt.Printf("    [%d](T=%d) 为 [%d](T=%d) 投票 YYYYY, 投票者log=%v, votefor=%d->%d\n",
				rf.me, prevT, args.CandidateId, args.Term, rf.log, prevVote, rf.votedFor)
		} else {
			fmt.Printf("    [%d](T=%d) 为 [%d](T=%d) 投票 NNNNN, 投票者log=%v, votefor=%d->%d\n",
				rf.me, prevT, args.CandidateId, args.Term, rf.log, prevVote, rf.votedFor)
		}
	}()

	reply.Term = rf.currentTerm
	reply.VoteGranted = false
	if rf.currentTerm > args.Term {
		return
	}

	defer rf.persist() // 2C

	if rf.currentTerm < args.Term {
		rf.votedFor = -1
		rf.currentTerm = args.Term
	}

	if (rf.votedFor == -1 || rf.votedFor == args.CandidateId) &&
		(rf.log[len(rf.log)-1].Term < args.LastLogTerm ||
			(rf.log[len(rf.log)-1].Term == args.LastLogTerm && len(rf.log)-1 <= args.LastLogIndex)) {
		reply.VoteGranted = true
		rf.votedFor = args.CandidateId
		rf.currentTerm = args.Term
		if rf.role != Follower {
			rf.role = Follower
			rf.transfer <- Follower
		}
	}

}

type AppendEntriesArgs struct {
	Term         int
	LeaderId     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []Log
	LeaderCommit int
}

type AppendEntriesReply struct {
	Term    int
	Success bool
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if rf.killed() {
		return
	}

	reply.Term = rf.currentTerm
	reply.Success = false
	if rf.currentTerm > args.Term {
		return
	}

	defer func() {
		rf.checkFollowerCommittedIndex(args, reply)
		rf.applyLog(args)
		rf.persist() // 2C
		if rf.role == Follower {
			rf.appendEvent <- Follower
		} else {
			rf.role = Follower
			rf.transfer <- Follower
		}

		var print bool = true
		for i := 0; i < len(rf.log); i++ {
			_, ok := rf.log[i].Command.(int)
			if !ok {
				print = false
				return
			}
		}
		if args.Entries == nil || len(args.Entries) == 0 {
			fmt.Printf("【收到心跳完毕】 [%d], currentTerm=%d, log=%v || total=%v, args.PrevLogIndex=%d, args.Term=%d, args.PrevLogTerm=%d %v  ===\n",
				rf.me, rf.currentTerm, rf.log, len(args.Entries)+args.PrevLogIndex+1, args.PrevLogIndex, args.Term, args.PrevLogTerm, time.Now())
			return
		}
		if print {
			fmt.Printf("【追加日志完毕】 [%d], currentTerm=%d, log=%v || total=%v, args.PrevLogIndex=%d, args.Term=%d, args.PrevLogTerm=%d, args.LeaderCommit=%d, entries=%v %v ===\n",
				rf.me, rf.currentTerm, rf.log, len(args.Entries)+args.PrevLogIndex+1, args.PrevLogIndex, args.Term, args.PrevLogTerm, args.LeaderCommit, args.Entries, time.Now())
		} else {
			fmt.Printf("【追加日志完毕】 日志含有字符串 [%d], currentTerm=%d || total=%v, args.PrevLogIndex=%d, args.Term=%d, args.LeaderCommit=%d, args.PrevLogTerm=%d .... ===\n",
				rf.me, rf.currentTerm, len(args.Entries)+args.PrevLogIndex+1, args.PrevLogIndex, args.Term, args.LeaderCommit, args.PrevLogTerm)
		}
	}()

	if rf.currentTerm < args.Term {
		rf.currentTerm = args.Term
		rf.votedFor = -1
	}

	if args.Entries == nil || len(args.Entries) == 0 {

		return
	}

	if len(rf.log)-1 < args.PrevLogIndex || rf.log[args.PrevLogIndex].Term != args.PrevLogTerm {
		return
	}

	reply.Success = true
	var index int
	for i := 0; i < len(args.Entries); i++ {
		index = args.PrevLogIndex + i + 1
		if index < len(rf.log) {

			rf.log[index] = args.Entries[i]
		} else {
			rf.log = append(rf.log, args.Entries[i])
		}
	}
	rf.log = rf.log[:len(args.Entries)+args.PrevLogIndex+1]
	rf.committedIndex = minIntValue(len(rf.log)-1, args.LeaderCommit)
}

func (rf *Raft) checkFollowerCommittedIndex(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	if len(rf.log) > args.PrevLogIndex && rf.log[args.PrevLogIndex].Term == args.PrevLogTerm {
		rf.committedIndex = minIntValue(maxIntValue(rf.committedIndex, args.PrevLogIndex), args.LeaderCommit)
	}
}

func (rf *Raft) applyLog(args *AppendEntriesArgs) {
	for i := rf.lastApplied + 1; i <= rf.committedIndex && i < len(rf.log); i++ {
		rf.lastApplied = i
		msg := ApplyMsg{
			CommandValid: true,
			CommandIndex: i,
			Command:      rf.log[i].Command,
		}
		rf.ApplyChan <- msg
		// fmt.Printf("$$$ [%d] Follower applyIndex=%d, val=%v $$$\n", rf.me, rf.lastApplied, msg.Command)
	}
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	// Your code here (2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	defer rf.persist()
	if rf.role == Leader {

		index = len(rf.log)
		term = rf.currentTerm
		isLeader = true

		l := Log{Command: command, Term: rf.currentTerm}
		rf.log = append(rf.log, l)
		rf.jobSignal <- Leader
		_, ok := command.(int)
		if ok {
			fmt.Printf("【New Start】==== index=%d, term=%d, leaderId=%d commitIndex=%d, log=%d =====\n",
				index, term, rf.me, rf.committedIndex, rf.log)
		}
	} else {
		isLeader = false
	}

	return index, term, isLeader
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
	// fmt.Printf("            magic【%d killed】: %v \n", rf.me, time.Now())

	// rf.destory()
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int, persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).
	rf.ApplyChan = applyCh
	rf.currentTerm = 0
	rf.votedFor = -1
	rf.matchIndex = make([]int, len(rf.peers))
	rf.nextIndex = make([]int, len(rf.peers))

	for i := 0; i < len(rf.peers); i++ {
		rf.nextIndex[i] = 1
	}
	rf.log = make([]Log, 0)
	rf.log = append(rf.log, Log{Command: -1, Term: rf.currentTerm})

	rf.initResource()

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	fmt.Printf("            ================= Make [%d] ================      \n", rf.me)
	go rf.followerLoop()

	return rf
}

// leader 需要周期性发送 heartbeat 维持 leader 状态
func (rf *Raft) leaderLoop() {
	fmt.Printf("[leaderLoop....] leader[%d]=(%d)     %v\n", rf.me, rf.currentTerm, time.Now())
	rf.role = Leader
	go rf.doLeaderJob()
	for {
		if rf.killed() {
			return
		}

		select {
		case v, ok := <-rf.transfer:
			if ok && v == Follower {
				rf.jobSignal <- EXIT
				fmt.Printf("[leaderLoop....]=====> rf.transfer, [%d](%d)    %v\n", rf.me, rf.currentTerm, time.Now())
				go rf.followerLoop()
				return
			}
		case v, ok := <-rf.voteEvent:
			if ok && v == Follower {
				rf.jobSignal <- EXIT
				fmt.Printf("[leaderLoop....]=====> rf.voteEvent, [%d](%d)    %v\n", rf.me, rf.currentTerm, time.Now())
				go rf.followerLoop()
				return
			}
		case v, ok := <-rf.appendEvent:
			if ok && v == Follower {
				rf.jobSignal <- EXIT
				fmt.Printf("[leaderLoop....]=====> rf.appendEvent, [%d](%d)    %v\n", rf.me, rf.currentTerm, time.Now())
				go rf.followerLoop()
				return
			}
		}

	}
}

func (rf *Raft) doLeaderJob() {
	for {
		if rf.killed() {
			return
		}
		for i := 0; i < len(rf.peers); i++ {
			if i == rf.me {
				continue
			}

			rf.mu.Lock()
			rf.checkCommittedIndex()
			rf.applyLeaderLog()
			rf.mu.Unlock()

			if len(rf.log) != rf.nextIndex[i] && tryLock(&rf.members[i]) {
				go rf.inconsistentLogHandler(i)
			} else {
				go rf.sendHeartbeat(i)
			}

		}
		select {
		case <-time.After(130 * time.Millisecond):
			break
		case x, ok := <-rf.jobSignal:
			if ok && x == Leader {
				break
			} else {
				return
			}
		}
	}
}

func (rf *Raft) sendHeartbeat(server int) {
	rf.mu.Lock()
	if rf.nextIndex[server] > len(rf.log) {
		rf.mu.Unlock()
		return
	}
	args := &AppendEntriesArgs{
		Term:         rf.currentTerm,
		LeaderId:     rf.me,
		PrevLogIndex: rf.nextIndex[server] - 1,
		PrevLogTerm:  rf.log[rf.nextIndex[server]-1].Term,
		LeaderCommit: rf.committedIndex,
	}
	reply := &AppendEntriesReply{}
	rf.mu.Unlock()

	if rf.sendAppendEntries(server, args, reply) {

		rf.mu.Lock()

		if rf.currentTerm < reply.Term {
			rf.currentTerm = reply.Term
			rf.votedFor = -1
			fmt.Printf("【transfer sendHeartbeat 】[%d] Leader => Follower ... %v \n", rf.me, time.Now())
			rf.transfer <- Follower
		}
		rf.mu.Unlock()
	}
}

func (rf *Raft) checkCommittedIndex() {
	// var threshold int = len(rf.peers)/2 + 1
	// for {
	// 	var counter int = 0
	// 	for i := 0; i < len(rf.peers); i++ {
	// 		if i == rf.me || rf.nextIndex[i] > rf.committedIndex+1 {
	// 			counter++
	// 		}
	// 		// if i == rf.me || rf.matchIndex[i] > rf.committedIndex {
	// 		// 	counter++
	// 		// }
	// 	}
	// 	if counter < threshold {
	// 		return
	// 	}
	// 	rf.committedIndex++
	// }

	var threshold int = len(rf.peers) / 2
	for i := rf.committedIndex + 1; i < len(rf.log); i++ {
		var counter int = 0
		for j := 0; j < len(rf.peers); j++ {
			if j == rf.me || (rf.matchLog[j].term == rf.currentTerm && rf.matchLog[j].beg <= i && rf.matchLog[j].end >= i) {
				counter++
			}
			if counter > threshold {
				rf.committedIndex = i
			}
		}
	}
}

func (rf *Raft) applyLeaderLog() {
	var threshold int = len(rf.peers)/2 + 1
	for i := rf.lastApplied + 1; i <= rf.committedIndex; i++ {
		var counter int = 1
		for j := 0; j < len(rf.peers); j++ {
			if j == rf.me {
				continue
			}
			if rf.nextIndex[j] > i {
				counter++
			}
		}
		if counter < threshold {
			break
		}
		msg := ApplyMsg{
			CommandValid: true,
			CommandIndex: i,
			Command:      rf.log[i].Command,
		}
		rf.ApplyChan <- msg
		rf.lastApplied = i
		// if _, ok := rf.log[i].Command.(int); ok {
		// 	fmt.Printf("[%d]   Leader applyIndex=%d, val=%d\n", rf.me, rf.lastApplied, rf.log[i].Command)
		// }
	}
}

func (rf *Raft) inconsistentLogHandler(server int) {
	defer rf.members[server].Unlock()

	if rf.role != Leader {
		return
	}
	if len(rf.log) == rf.nextIndex[server] {
		return
	}
	// var done chan interface{} = make(chan interface{}, 0)
	go func() {
		rf.mu.Lock()
		if rf.nextIndex[server] > len(rf.log) {
			rf.mu.Unlock()
			return
		}
		entries := rf.log[rf.nextIndex[server]:len(rf.log)]
		args := AppendEntriesArgs{
			Term:         rf.currentTerm,
			LeaderId:     rf.me,
			PrevLogIndex: rf.nextIndex[server] - 1,
			PrevLogTerm:  rf.log[rf.nextIndex[server]-1].Term,
			Entries:      entries,
			LeaderCommit: rf.committedIndex,
		}
		reply := AppendEntriesReply{}
		rf.mu.Unlock()

		if rf.sendAppendEntries(server, &args, &reply) {
			if rf.currentTerm != args.Term {
				return
			}
			rf.mu.Lock()
			defer rf.mu.Unlock()

			if rf.role != Leader ||
				rf.currentTerm != args.Term ||
				rf.nextIndex[server]-1 != args.PrevLogIndex ||
				rf.log[rf.nextIndex[server]-1].Term != args.PrevLogTerm {

				return
			}
			if reply.Success {
				index := len(args.Entries) + args.PrevLogIndex + 1
				rf.nextIndex[server] = index
				rf.matchIndex[server] = index - 1
				rf.backup[server] = 1

				if rf.matchLog[server].term != rf.currentTerm {
					rf.matchLog[server].term = rf.currentTerm
					rf.matchLog[server].beg = 9999999999
					rf.matchLog[server].end = -1
				}
				for i := args.PrevLogIndex + 1; i < index; i++ {
					if rf.log[i].Term == rf.currentTerm {
						rf.matchLog[server].beg = minIntValue(rf.matchLog[server].beg, i)
						rf.matchLog[server].end = maxIntValue(rf.matchLog[server].end, i)
					}
				}

			} else if rf.currentTerm >= reply.Term {
				// rf.nextIndex[server]--
				rf.nextIndex[server] = maxIntValue(rf.nextIndex[server]-rf.backup[server], 1)
				rf.backup[server] = minIntValue(len(rf.log), rf.backup[server]*2)
			} else if rf.currentTerm < reply.Term {
				rf.votedFor = -1
				rf.currentTerm = reply.Term
				rf.role = Follower
				fmt.Printf("【transfer inconsistentLogHandler】[%d] Leader => Follower ... %v \n", rf.me, time.Now())
				rf.transfer <- Follower
			}
			rf.persist() // 2C

		}
	}()
	select {
	case <-time.After(time.Duration(500) * time.Millisecond):
		return
	}
}

func (rf *Raft) followerLoop() {
	rf.role = Follower
	var dur int64 = timeout()
	fmt.Printf("in followerLoop(me=%d, currentTerm=%d), 超时时间=%v\n", rf.me, rf.currentTerm, dur)
	for {
		if rf.killed() {
			return
		}
		select {
		case <-rf.voteEvent:
		case <-rf.appendEvent:
		case <-time.After(time.Duration(dur) * time.Millisecond):
			fmt.Printf("       followerLoop 等待超时 (me=%d, currentTerm=%d), 超时时间=%v\n", rf.me, rf.currentTerm, dur)
			go rf.candidateLoop()
			return
		}

	}
}

func (rf *Raft) candidateLoop() {
	fmt.Printf("in candidateLoop(me=%d, currentTerm=%d) \n", rf.me, rf.currentTerm)
	rf.role = Candidate

	go rf.doCandidateJob()
	if rf.killed() {
		return
	}
	var r Role
	var ok bool
	select {
	case r, ok = <-rf.transfer:
		if !ok {
			return
		}
		if r == Leader && rf.role == Leader {
			go rf.leaderLoop()
		} else if r == Follower && rf.role == Follower {
			go rf.followerLoop()
		}
		return
	case r, ok = <-rf.appendEvent:
		if ok && r == Follower && rf.role == Follower {
			go rf.followerLoop()
		}
		return
	case r, ok = <-rf.voteEvent:
		if ok && r == Follower && rf.role == Follower {
			go rf.followerLoop()
		}
		return
	}

}

func (rf *Raft) doCandidateJob() {
	var done chan Role = make(chan Role, 0)
	defer func() {
		close(done)
	}()
	for {
		if rf.killed() {
			return
		}
		if rf.role != Candidate {
			return
		}

		rf.currentTerm++
		rf.votedFor = rf.me
		rf.persist() // 2C

		var size int = len(rf.peers)
		argss := make([]RequestVoteArgs, size)
		replys := make([]RequestVoteReply, size)
		for i := 0; i < size; i++ {
			if i == rf.me {
				continue
			}
			argss[i] = RequestVoteArgs{
				Term:         rf.currentTerm,
				CandidateId:  rf.me,
				LastLogTerm:  rf.log[len(rf.log)-1].Term,
				LastLogIndex: len(rf.log) - 1,
			}
			replys[i] = RequestVoteReply{}
		}

		var granted int32 = 0
		go func() {
			for i := 0; i < size; i++ {
				if rf.me == i {
					atomic.AddInt32(&granted, 1)
					continue
				}
				go func(server int, args *RequestVoteArgs, reply *RequestVoteReply) {
					fmt.Printf("   == [%d](%d) 请求 [%d] 投票, LastLogTerm=%d, LastLogIndex=%d  %v\n",
						rf.me, rf.currentTerm, server, rf.log[len(rf.log)-1].Term, len(rf.log)-1, time.Now())
					if rf.sendRequestVote(server, args, reply) {
						rf.mu.Lock()
						defer rf.mu.Unlock()
						if rf.role != Candidate ||
							rf.currentTerm > args.Term ||
							rf.votedFor != rf.me {
							return
						}
						if reply.VoteGranted {
							atomic.AddInt32(&granted, 1)
						}
						if granted > (int32)(size/2) {
							// fmt.Printf("before rf.jobSigna l ->【leader】 [%d](%d)   %v\n", rf.me, rf.currentTerm, time.Now())
							rf.role = Leader
							// rf.jobSignal <- Leader
							// fmt.Printf("after   rf.jobSignal ->【leader】 [%d](%d)   %v\n", rf.me, rf.currentTerm, time.Now())
							done <- Leader
							return
						}
						if rf.currentTerm < reply.Term {
							rf.role = Follower
							rf.currentTerm = reply.Term
							rf.votedFor = -1
							rf.persist() // 2C
							// fmt.Printf("   rf.jobSignal ->【Follower】 [%d](%d)   %v\n", rf.me, rf.currentTerm, time.Now())
							// rf.jobSignal <- Follower
							done <- Follower
						}
					}
				}(i, &argss[i], &replys[i])
			}
		}()

		select {
		case <-time.After(time.Duration(timeout()) * time.Millisecond):
			// fmt.Printf("eclapse catch rf.jobSignal ->【%s】 [%d](%d)   %v\n", roleMsg[rf.role], rf.me, rf.currentTerm, time.Now())
			if rf.role == Leader {
				rf.transfer <- Leader
				return
			}
			break
		case y, ok := <-rf.jobSignal:
			// fmt.Printf("before catch rf.jobSignal ->【%s】 [%d](%d)   %v\n", roleMsg[rf.role], rf.me, rf.currentTerm, time.Now())
			// if ok  {
			// 	fmt.Printf("after catch rf.jobSignal ->【%s】 [%d](%d)   %v\n", roleMsg[rf.role], rf.me, rf.currentTerm, time.Now())
			// 	rf.transfer <- y
			// }
			// return
			if ok && y == EXIT {
				return
			}
		case r, ok := <-done:
			if ok {
				rf.transfer <- r
				return
			}
		}
	}
}

//  === helper function ===

func (rf *Raft) initResource() {
	rf.transfer = make(chan Role, 102400)
	rf.jobSignal = make(chan Role, 102400)
	rf.voteEvent = make(chan Role, 102400)
	rf.appendEvent = make(chan Role, 102400)
	// rf.clientLog = make(chan Log)
	rf.backup = make([]int, len(rf.peers))
	rf.members = make([]sync.Mutex, len(rf.peers))

	rf.matchLog = make([]MatchLog, len(rf.peers))

	for i := 0; i < len(rf.peers); i++ {
		rf.members[i] = sync.Mutex{}
		rf.backup[i] = 1
		rf.matchLog[i] = MatchLog{
			beg:  99999999,
			end:  -1,
			term: -1,
		}
	}
}

func (rf *Raft) freeResource() {

}

// paper : election timeout = [150, 300)
// actual: election timeout = [450, 900)
func timeout() int64 {
	max := big.NewInt(450)
	rr, _ := crand.Int(crand.Reader, max)
	t := rr.Int64() + 450
	return t
}

func minIntValue(i, j int) int {
	if i < j {
		return i
	}
	return j
}

func maxIntValue(i, j int) int {
	if i < j {
		return j
	}
	return i
}

func withoutPanicCloseChan(C chan Role) bool {
	var res bool = true
	defer func() {
		if recover() != nil {
			res = false
		}
	}()
	close(C)
	return res
}

func withoutPanicCloseChan2(C chan interface{}) bool {
	var res bool = true
	defer func() {
		if recover() != nil {
			res = false
		}
	}()
	close(C)
	return res
}

func withoutPanicSendToChan(c chan Role, v Role) bool {
	var res bool = true
	defer func() {
		if recover() != nil {
			res = false
		}
	}()
	c <- v
	return res
}

func withoutPanicSendToChan2(c chan interface{}, v interface{}) bool {
	var res bool = true
	defer func() {
		if recover() != nil {
			res = false
		}
	}()
	c <- v
	return res
}

func closeAfterWrite(c chan Role, v Role) {
	defer func() {
		if recover() != nil {

		}
	}()
	c <- v
	close(c)
}

const mutexLocked = 1 << iota

func tryLock(m *sync.Mutex) bool {
	return atomic.CompareAndSwapInt32((*int32)(unsafe.Pointer(m)), 0, mutexLocked)
}
